// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: linchuan

#include "coin2/exchange/kraken/order_v1/parser.h"

#include <memory>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>

#include <rapidjson/document.h>

#include "coin/proto/coin_query.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/base/symbology/util.h"
#include "coin2/exchange/kraken/symbology/product_info.h"
#include "coin2/exchange/util/json_util.h"
namespace coin2::exchange::kraken::order_v1 {

namespace {

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;

using base::order::IOrderContext;
using symbology::KrakenCurrency;
using symbology::KrakenProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;

inline int32_t ProcOrderIdToUserRef(int64_t proc_order_id) {
  int32_t* x = reinterpret_cast<int32_t*>(&proc_order_id);
  return x[0] + x[1];
}

inline bool IsClosedOrder(const JsonValue& data) {
  if (!data.HasMember("status")) {
    return false;
  }
  std::string_view status = data["status"].GetString();
  if (status == "closed" || status == "canceled" || status == "expired") {
    return true;
  }
  return false;
}

template <typename T>
inline const char* GetUserRef(const T& value) {
  static char number[128];
  if (value.HasMember("userref")) {
    if (value["userref"].IsString()) {
      return value["userref"].GetString();
    } else if (value["userref"].IsInt64()) {
      snprintf(number, sizeof(number), "%ld", value["userref"].GetInt64());
      return number;
    }
  }
  return "0";
}

inline auto GetNativeSymbolFromAltName(const std::string& alt_name) -> std::optional<std::string> {
  static const base::symbology::ProductInfoConstant kKrakenProductInfo(
      "data/coin2/product_info/Spot.Kraken.v1.json");
  static std::unordered_map<std::string, std::string> alt_name_map = [&]() -> auto{
    std::unordered_map<std::string, std::string> ret;
    const auto& pis = kKrakenProductInfo.GetProductInfoBundle();
    for (const auto& pi : pis.product_infos()) {
      if (!pi.has_altname()) {
        continue;
      }
      if (pi.altname() == pi.native_symbol()) {
        continue;
      }
      ret.emplace(pi.altname(), pi.native_symbol());
    }

    return ret;
  }
  ();
  if (!alt_name_map.count(alt_name)) {
    return {};
  }
  return alt_name_map.at(alt_name);
}

inline std::string ToNativeProductName(const std::string& symbol) {
  if (auto ret = GetNativeSymbolFromAltName(symbol)) {
    return *ret;
  }

  return symbol;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "buy") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "sell") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "limit") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "market") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

inline void SetOrderStatus(const JsonValue& data, ProductOrderElement* order) {
  std::string status = data["status"].GetString();
  if (status == "closed" || status == "expired") {
    order->set_is_live(false);
  } else if (status == "open") {
    order->set_is_live(true);
  } else if (status == "canceled") {
    order->set_cancel_confirmed(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
}

inline bool IsKrakenErrMessage(const JsonValue& doc) {
  // {"error":["EAPI:Invalid key"]}
  return doc.HasMember("error") && doc["error"].IsArray() && (doc["error"].Size() > 0);
}

inline std::string GetKrakenExchangeOrderErrorMsg(const JsonValue& json) {
  return json["error"][0].GetString();
}

inline coin::proto::OrderErrorCode GetOrderErrorCodeFromKraken(const JsonValue& json) {
  using coin::proto::OrderErrorCode;
  auto result = coin::proto::OrderErrorCode::ORDER_UNKNOWN_ERROR;
  std::string error_code(json["error"][0].GetString());

  if (error_code == "0") {
    result = OrderErrorCode::ORDER_NO_ERROR;
  } else if (error_code == "EOrder:Unknown order") {
    result = OrderErrorCode::ORDER_NOT_FOUND;
  } else if (error_code == "EOrder:Insufficient funds") {
    result = OrderErrorCode::SUBMIT_ORDER_INSUFFICIENT_FUND;
  } else if (error_code == "EGeneral:Invalid arguments:volume") {
    result = OrderErrorCode::SUBMIT_ORDER_WRONG_QTY;
  } else {
    LOG(ERROR) << "Unknown error code: " << error_code;
  }

  return result;
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  if (!json.IsArray()) {
    return false;
  }
  if (json.Size() < 2) {
    return false;
  }
  std::string ws_type = json[1].GetString();
  if (ws_type != "openOrders" && ws_type != "ownTrades") {
    return false;
  }
  return true;
}

}  // namespace

void KrakenParser::ParseRestAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  auto account_balance = ParseBalance(document);
  if (!account_balance.has_exchange() || !account_balance.has_market_type()) {
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

// NOTE: when field exchange and market_type are set, which means parsed successfully
AccountBalance KrakenParser::ParseBalance(const JsonValue& doc) {
  RawBalanceMap raw_balance_map;
  auto success = ParseRawBalance(doc, &raw_balance_map);
  if (!success) {
    return {};
  }

  AccountBalance acc_balance;
  acc_balance.set_exchange("Kraken");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (raw_balance_map.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(raw_balance_map.at(native_currency));
    }
    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available();
  }
  is_account_balance_initialized_ = true;

  return acc_balance;
}

bool KrakenParser::ParseRawBalance(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "error": [],
    "result": {
      "XXBT": "1.4399999967",
      "XXRP": "0.00000000",
      "XETH": "58.7646926300",
      "EOS": "0.0000542300",
      "BSV": "0.0000000000"
    }
  }
  */
  if (doc.IsObject() && IsKrakenErrMessage(doc)) {
    LOG(ERROR) << "[ParseBalance] " << JsonToString(doc);
    return false;
  }
  if (!doc.IsObject() || !doc.HasMember("result") || !doc["result"].IsObject()) {
    LOG(ERROR) << "[ParseBalance] unexpected balance format!" << JsonToString(doc);
    return false;
  }

  for (const auto& info : doc["result"].GetObject()) {
    CurrencyBalance currency_balance;
    currency_balance.set_currency_native(info.name.GetString());
    currency_balance.set_total(stod(info.value.GetString()));
    raw_balance_map->emplace(currency_balance.currency_native(), currency_balance);
  }

  return true;
}

ProductOrderElement KrakenParser::ParseRestOrderInfo(
    const std::string order_id,
    const JsonValue& data) {
  /*
  {
    "error": [],
    "result": {
      "open": {
        "OWTJ5T-RFFOH-5TYYAD": {
          "refid": null,
          "userref": 0,
          "status": "open",
          "opentm": 1599647607.5597,
          "starttm": 0,
          "expiretm": 0,
          "descr": {
            "pair": "BCHUSD",
            "type": "sell",
            "ordertype": "limit",
            "price": "225.1",
            "price2": "0",
            "leverage": "none",
            "order": "sell 3.30000000 BCHUSD @ limit 225.1",
            "close": ""
          },
          "vol": "3.30000000",
          "vol_exec": "0.00000000",
          "cost": "0.000000",
          "fee": "0.000000",
          "price": "0.000000",
          "stopprice": "0.000000",
          "limitprice": "0.000000",
          "misc": "",
          "oflags": "fciq,post"
        },
        "O5KU7F-5L63B-7FRCPH": {
          ...
        }
      }
    }
  }
  */
  ProductOrderElement order;
  order.set_market_type(MarketType::Spot);
  order.set_exchange_type(ExchangeType::Kraken);
  order.set_api_version("v1");
  order.set_exchange_order_id(order_id);
  auto oc = FindOrderContext(order_id, GetUserRef(data));
  if (oc != nullptr) {
    order.set_proc_order_id(oc->proc_order_id());
  } else {
    if (!IsClosedOrder(data)) {
      LOG(INFO) << "userref not found!" << JsonToString(data);
    }
  }
  order.set_symbol(ToNativeProductName(data["descr"]["pair"].GetString()));
  order.set_price(stod(data["descr"]["price"].GetString()));
  order.set_qty(stod(data["vol"].GetString()));
  order.set_side(GetOrderSide(data["descr"]["type"].GetString()));
  order.set_order_type(GetOrderType(data["descr"]["ordertype"].GetString()));
  order.set_timestamp(msg_->timestamp);
  SetOrderStatus(data, &order);
  return order;
}

void KrakenParser::ParseRestOrdersInfo() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOrdersInfo] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsKrakenErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrdersInfo] " << JsonToString(data);
    return;
  }

  if (!data.IsObject() || !data.HasMember("result") || !data["result"].IsObject()) {
    LOG(ERROR) << "[ParseRestOrdersInfo] Unexpected order list format!" << JsonToString(data);
    return;
  }

  const char* field = nullptr;
  if (data["result"].HasMember("open")) {
    field = "open";
  } else if (data["result"].HasMember("closed")) {
    field = "closed";
  } else {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();
  bool product_order_bundle_found = false;
  for (const auto& info : data["result"][field].GetObject()) {
    std::string name = info.name.GetString();
    const auto& value = info.value.GetObject();
    if (!value.HasMember("status") || !value.HasMember("vol") || !value.HasMember("descr") ||
        !value["descr"].HasMember("price") || !value["descr"].HasMember("pair") ||
        !value["descr"].HasMember("type") || !value["descr"].HasMember("ordertype") ||
        !value["status"].IsString() || !value["vol"].IsString() ||
        !value["descr"]["price"].IsString() || !value["descr"]["pair"].IsString() ||
        !value["descr"]["type"].IsString() || !value["descr"]["ordertype"].IsString()) {
      LOG(ERROR) << "Parse missing field:\n" << JsonToString(info.value);
      continue;
    }
    if (std::string_view(value["status"].GetString()) == "pending") {
      continue;
    }
    auto* order = product_order_bundle->add_each_order();
    auto new_order = ParseRestOrderInfo(name, info.value);
    order->MergeFrom(new_order);
    product_order_bundle_found = true;
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle KrakenParser::ParseWsOrderInfoListV2(const JsonValue& data) {
  /*
  [
    [
      {
        "OUULIP-WZKEY-MOWRI4": {
          "avg_price": "0.000000",
          "cost": "0.000000",
          "descr": {
            "close": null,
            "leverage": null,
            "order": "buy 3.01000000 EOS/USD @ limit 2.651000",
            "ordertype": "limit",
            "pair": "EOS/USD",
            "price": "2.651000",
            "price2": "0.000000",
            "type": "buy"
          },
          "expiretm": null,
          "fee": "0.000000",
          "limitprice": "0.000000",
          "misc": "",
          "oflags": "fciq,post",
          "opentm": "1600235339.783493",
          "refid": null,
          "starttm": null,
          "status": "pending",
          "stopprice": "0.000000",
          "userref": "0",
          "vol": "3.01000000",
          "vol_exec": "0.00000000"
        }
      }
    ],
    "openOrders"
  ]
  */
  ProductOrderBundle product_order_bundle;

  ratecount_map_.clear();
  for (const auto& data_arr : data.GetArray()) {
    for (const auto& info : data_arr.GetObject()) {
      std::string name = info.name.GetString();
      const auto& value = info.value.GetObject();

      if (!value.HasMember("status")) {
        continue;
      }
      if (std::string_view(value["status"].GetString()) == "pending") {
        ProductOrderElement order;
        std::string symbol = value["descr"]["pair"].GetString();
        boost::replace_all(symbol, "/", "");
        symbol = ToNativeProductName(symbol);
        order.set_market_type(MarketType::Spot);
        order.set_exchange_type(ExchangeType::Kraken);
        auto oc = FindOrderContext(name, GetUserRef(value));
        if (oc != nullptr) {
          order.set_proc_order_id(oc->proc_order_id());
        } else {
          LOG(INFO) << "userref not found!" << JsonToString(data);
        }
        order.set_api_version("v1");
        order.set_exchange_order_id(name);
        order.set_symbol(symbol);
        order.set_price(stod(value["descr"]["price"].GetString()));
        order.set_qty(stod(value["vol"].GetString()));
        order.set_side(GetOrderSide(value["descr"]["type"].GetString()));
        order.set_order_type(GetOrderType(value["descr"]["ordertype"].GetString()));
        if (ws_order_map_.find(name) == ws_order_map_.end()) {
          ws_order_map_.insert(std::make_pair(name, order));
        } else {
          LOG(INFO) << "Duplicate pending ws order: " << JsonToString(data_arr);
          ws_order_map_[name.c_str()] = order;
        }
      } else {
        if (ws_order_map_.find(name) == ws_order_map_.end()) {
          LOG(INFO) << "No pending msg: " << JsonToString(data_arr);
        } else {
          auto* order = product_order_bundle.add_each_order();
          order->MergeFrom(ws_order_map_[name.c_str()]);
          order->set_timestamp(msg_->timestamp);
          SetOrderStatus(info.value, order);
        }
      }
      if (value.HasMember("ratecount")) {
        const auto rate_count = std::stoi(value["ratecount"].GetString());
        if (rate_count > 100) {  // do not update rate limit so frequently
          if (auto* oc = ocm_->FindByExchangeOrderId(name); oc != nullptr) {
            const auto& symbol = oc->native_product();
            ratecount_map_[symbol] = rate_count;
          }
        }
      }
    }
  }

  return product_order_bundle;
}

void KrakenParser::ParseRestFillInfoList() {
  /*
  {
    "error": [],
    "result": {
      "trades": {
        "T5VZJJ-MUPID-TBDGZK": {
          "ordertxid": "OG6W5X-2Q7OC-KNCBQS",
          "postxid": "TKH2SE-M7IF5-CFI7LT",
          "pair": "XETHZUSD",
          "time": 1599644541.9439,
          "type": "sell",
          "ordertype": "limit",
          "price": "346.50000",
          "cost": "2338.87500",
          "fee": "2.33887",
          "vol": "6.75000000",
          "margin": "0.00000",
          "misc": ""
        },
        "TXNEQB-5LF2M-GZ4OMZ": {
          ...
        }
      }
    }
  }
  */
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsKrakenErrMessage(data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] " << JsonToString(data);
    return;
  }

  if (!data.IsObject() || !data.HasMember("result") || !data["result"].IsObject() ||
      !data["result"].HasMember("trades") || !data["result"]["trades"].IsObject()) {
    LOG(ERROR) << "Unexpected fill list format!" << JsonToString(data);
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();
  bool product_fill_bundle_found = false;
  for (const auto& info : data["result"]["trades"].GetObject()) {
    std::string name = info.name.GetString();
    const auto& value = info.value.GetObject();

    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(ToNativeProductName(value["pair"].GetString()));
    fill->set_fill_timestamp(msg_->timestamp);
    fill->set_exchange_order_id(value["ordertxid"].GetString());
    fill->set_price(stod(value["price"].GetString()));
    fill->set_qty(stod(value["vol"].GetString()));
    fill->set_fill_id(name);
    fill->set_fill_type(FillType::UNKNOWN_FILL_TYPE);
    fill->set_fee(stod(value["fee"].GetString()));
    product_fill_bundle_found = true;
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

ProductFillBundle KrakenParser::ParseWsFillInfoListV2(const JsonValue& data) {
  /*
  [
    [
      {
        "T2O5NS-JZBHU-FOBKYE": {
          "cost": "8.225126",
          "fee": "0.000000",
          "margin": "0.000000",
          "ordertxid": "OFEHX4-BZOTO-FHJI6B",
          "ordertype": "limit",
          "pair": "EOS/USD",
          "postxid": "TKH2SE-M7IF5-CFI7LT",
          "price": "2.732600",
          "time": "1600164543.914583",
          "type": "sell",
          "vol": "3.01000000"
        }
      },
      {
        "TEW535-ONSLZ-YCCNZV": {
          ......
        }
      }
    ],
    "ownTrades"
  ]
  */
  ProductFillBundle product_fill_bundle;

  for (const auto& data_arr : data.GetArray()) {
    for (const auto& info : data_arr.GetObject()) {
      std::string name = info.name.GetString();
      const auto& value = info.value.GetObject();

      auto* fill = product_fill_bundle.add_each_fill();
      std::string symbol = value["pair"].GetString();
      boost::replace_all(symbol, "/", "");
      symbol = ToNativeProductName(symbol);
      fill->set_symbol(ToNativeProductName(symbol));
      fill->set_fill_timestamp(msg_->timestamp);
      fill->set_exchange_order_id(value["ordertxid"].GetString());
      fill->set_price(stod(value["price"].GetString()));
      fill->set_qty(stod(value["vol"].GetString()));
      fill->set_fill_id(name);
      fill->set_fill_type(FillType::UNKNOWN_FILL_TYPE);
      fill->set_fee(stod(value["fee"].GetString()));
    }
  }
  return product_fill_bundle;
}

void KrakenParser::ParseSubmitResponse() {
  /*
  {
    "error": [],
    "result": {
      "descr": {
        "order": "sell 3.10000000 EOSUSD @ limit 3.3265"
      },
      "txid": ["OYIOKG-FLLGA-Q6PCIT"]
    }
  }
  */
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[KrakenSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[KrakenSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[KrakenSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!json.IsObject()) {
    LOG(ERROR) << "[KrakenSubmitOrder] Unexpected format\n" << http_context->res;
    resp->set_success(false);
    return;
  }

  if (IsKrakenErrMessage(json)) {
    auto error_code = GetOrderErrorCodeFromKraken(json);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetKrakenExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      if (error_code != coin::proto::OrderErrorCode::POST_ONLY_REJECT) {
        LOG(ERROR) << "[KrakenSubmitOrder] Error Response" << http_context->res;
      }
    }
    resp->set_success(false);
    return;
  }

  if (!json.HasMember("result") || !json["result"].HasMember("txid") ||
      !json["result"]["txid"].IsArray() || json["result"]["txid"].Size() == 0) {
    LOG(ERROR) << "[KrakenSubmitOrder] Unexpected format\n" << http_context->res;
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(json["result"]["txid"][0].GetString());
  std::string user_ref = std::to_string(ProcOrderIdToUserRef(order_context->proc_order_id()));
  if (oc_map_.count(user_ref) == 0) {
    oc_map_.insert(std::make_pair(user_ref, order_context));
  }
}

void KrakenParser::ParseCancelResponse() {
  // {"error":[],"result":{"count":1}}
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[KrakenCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  if (http_context->res.result_int() == 429) {
    LOG(ERROR) << "[KrakenCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[KrakenCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (!json.IsObject()) {
    LOG(ERROR) << "[KrakenCancelOrder] Unexpected format\n" << http_context->res;
    resp->set_success(false);
    return;
  }

  if (IsKrakenErrMessage(json)) {
    auto error_code = GetOrderErrorCodeFromKraken(json);
    resp->set_error_code(error_code);
    resp->set_exchange_error_msg(GetKrakenExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[KrakenCancelOrder] Error Response" << http_context->res;
    }
    resp->set_success(false);
    return;
  }

  if (!json.HasMember("result") || !json["result"].HasMember("count") ||
      std::string_view(json["result"]["count"].GetString()) != "1") {
    LOG(ERROR) << "[KrakenCancelOrder] Unexpected format\n" << http_context->res;
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);
}

void KrakenParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (IsOrderUpdate(document)) {
    std::string ws_type = document[1].GetString();
    if (ws_type == "ownTrades") {
      *result_.mutable_fill_list() = ParseWsFillInfoListV2(document[0]);
      result_.set_type(MsgType::ORDER_FILL);
    } else {
      *result_.mutable_order_list() = ParseWsOrderInfoListV2(document[0]);
      result_.set_type(MsgType::ORDER_STATUS);
      PublishRateCountUpdate();
    }
  } else {
    if (document.IsObject() && document.HasMember("channelName") &&
        std::string(document["channelName"].GetString()) == "openOrders") {
      if (document.HasMember("subscription")) {
        auto subscription = document["subscription"].GetObject();
        if (subscription.HasMember("maxratecount")) {
          LOG(INFO) << "[RateLimit] current max ratecount is "
                    << subscription["maxratecount"].GetString();
        }
      }
    }
  }
}

void KrakenParser::ParseListenKey() {
  // {"error":[],"result":{"token":"RkaBGj2CUYR032s1bY4QYbGVmVwQ\/rNuUVFqTn","expires":900}}
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseListenKey] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.HasMember("result") && document["result"].HasMember("token")) {
    std::string token(document["result"]["token"].GetString());
    mb_->PublishEvent("kraken/ws_token", &token);
  } else {
    LOG(ERROR) << "[ParseListenKey] failed to retrieve ws token";
    CHECK(false) << JsonToString(document);
  }
}

void KrakenParser::PublishRateCountUpdate() {
  if (ratecount_map_.size() == 0) {
    return;
  }
  nlohmann::json json(ratecount_map_);
  std::string json_str = json.dump();
  mb_->PublishEvent("kraken/order_rate_limit", reinterpret_cast<void*>(&json_str));
}

auto KrakenParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg.type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::LISTEN_KEY:
      ParseListenKey();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      // pass
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
    case OrderSubTopicId::REST_DONE_ORDERS:
      ParseRestOrdersInfo();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestAccountBalance();
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  return result_;
}

const IOrderContext* KrakenParser::FindOrderContext(
    const std::string& order_id,
    const std::string& user_ref) {
  if (user_ref == "0") {
    return nullptr;
  }

  if (oc_map_.count(user_ref) == 1) {
    return oc_map_.at(user_ref);
  }

  const int32_t user_ref_id = std::stoi(user_ref);
  for (auto* oc : ocm_->GetAllWorkingOrders()) {
    if (ProcOrderIdToUserRef(oc->proc_order_id()) == user_ref_id) {
      oc_map_.insert(std::make_pair(user_ref, oc));
      return oc;
    }
    if (!order_id.empty() && order_id == oc->exchange_order_id()) {
      auto user_ref = std::to_string(ProcOrderIdToUserRef(oc->proc_order_id()));
      oc_map_.insert(std::make_pair(user_ref, oc));
      return oc;
    }
  }

  return nullptr;
}

}  // namespace coin2::exchange::kraken::order_v1
